package DataStreamApi.Transformation算子;

import domain.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * 聚合算子，只能在KeyBy之后进行调用
 *
 *
 * MAX和MIN，取比较字段的最大值，非比较字段的初次值。
 *
 * MAXBY和MINBY，取最大值或者最小值的那条数据。
 *
 */

public class Flink05_max_min_sum_minby_maxby {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        List<WaterSensor> waterSensorList = Arrays.asList(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s2", 3L, 3),
                new WaterSensor("s3", 1L, 4),
                new WaterSensor("s1", 11L, 5),
                new WaterSensor("s1", 2L, 6));
        executionEnvironment.setParallelism(3);

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = executionEnvironment.fromCollection(waterSensorList);


        /**
         * 按照ID分组
         * KeyBY不是转换算子，只是对数据进行了重分区，不能设置并行度。
         *
         * 1、KeyBy是对数据进行分组，保证相同key的数据在同一个分区
         * 2、分区是指将数据划分到不同的区域，一个子任务可以认为是一个分区。
         *
         */
        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = waterSensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        });


        SingleOutputStreamOperator<WaterSensor> sum = waterSensorStringKeyedStream.sum("vc");

//        sum.print();

        SingleOutputStreamOperator<WaterSensor> min = waterSensorStringKeyedStream.min("vc");
//        min.print();

        SingleOutputStreamOperator<WaterSensor> max = waterSensorStringKeyedStream.max("vc");
//        max.print();



        SingleOutputStreamOperator<WaterSensor> maxBy = waterSensorStringKeyedStream.maxBy("vc");
//        max.print();

        SingleOutputStreamOperator<WaterSensor> minBy = waterSensorStringKeyedStream.minBy("vc");


        executionEnvironment.execute();

    }
}
